package com.dc.sbmq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitMQ
 * 注意：交换机、队列、绑定实例装配到Spring IOC容器，项目在启动的时候会在RabbitMQ服务器创建交换机、队列以及绑定关系
 *
 * @author duchao
 * @EnableRabbit 开启注解监听
 */
@Slf4j
@EnableRabbit
@Configuration
public class RabbitMQConfig {

    /**************************************************************
     * 直连交换机（绑定死信交换机）
     **************************************************************/
    public static final String DEMO_DIRECT_EXCHANGE = "demo-direct-exchange";
    public static final String DEMO_DIRECT_QUEUE = "demo-direct-queue";
    public static final String DEMO_DIRECT_ROUTING_KEY = "demo-routing-key";

    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(DEMO_DIRECT_EXCHANGE, true, false);
    }

    @Bean
    public Queue directQueue() {
        Queue queue = new Queue(DEMO_DIRECT_QUEUE, true);
        //为队列声明绑定死信交换机参数
        //queue.addArgument("x-dead-letter-exchange",DEMO_DEAD_LETTER_DIRECT_EXCHANGE);
        //queue.addArgument("x-dead-letter-routing-key",DEMO_DEAD_LETTER_ROUTING_KEY1);
        //queue.addArgument("x-message-ttl", 2000000);
        return queue;
    }

    @Bean
    public Binding directBinding() {
        return BindingBuilder.bind(directQueue()).to(directExchange()).with(DEMO_DIRECT_ROUTING_KEY);
    }

    /**************************************************************
     * 直连交换机TTL+死信队列 ==> 延迟队列
     **************************************************************/
    public static final String DEMO_DELAY_IN_EXCHANGE = "demo-delay-in-exchange";
    public static final String DEMO_DELAY_OUT_EXCHANGE = "demo-delay-out-exchange";
    public static final String DEMO_DELAY_IN_QUEUE = "demo-delay-in-queue";
    public static final String DEMO_DELAY_OUT_QUEUE = "demo-delay-out-queue";
    public static final String DEMO_DELAY_IN_ROUTING_KEY = "demo.delay.routing.key.in";
    public static final String DEMO_DELAY_OUT_ROUTING_KEY = "demo.delay.routing.key.out";

    @Bean
    public DirectExchange delayInExchange() {
        return new DirectExchange(DEMO_DELAY_IN_EXCHANGE, true, false);
    }

    @Bean
    public Queue delayInQueue() {
        Queue queue = new Queue(DEMO_DELAY_IN_QUEUE, true);
        //为队列声明绑定死信交换机参数
        queue.addArgument("x-dead-letter-exchange",DEMO_DELAY_OUT_EXCHANGE);
        queue.addArgument("x-dead-letter-routing-key",DEMO_DELAY_OUT_ROUTING_KEY);
        queue.addArgument("x-message-ttl", 1000);
        return queue;
    }

    @Bean
    public Binding delayInBinding() {
        return BindingBuilder.bind(delayInQueue()).to(delayInExchange()).with(DEMO_DELAY_IN_ROUTING_KEY);
    }

    @Bean
    public DirectExchange delayOutExchange() {
        return new DirectExchange(DEMO_DELAY_OUT_EXCHANGE, true, false);
    }

    @Bean
    public Queue delayOutQueue() {
        return new Queue(DEMO_DELAY_OUT_QUEUE, true);
    }

    @Bean
    public Binding delayOutBinding() {
        return BindingBuilder.bind(delayOutQueue()).to(delayOutExchange()).with(DEMO_DELAY_OUT_ROUTING_KEY);
    }

    /**************************************************************
     * 主题交换机
     **************************************************************/
    public static final String DEMO_TOPIC_EXCHANGE = "demo-topic-exchange";
    public static final String DEMO_TOPIC_QUEUE1 = "demo-topic-queue1";
    public static final String DEMO_TOPIC_QUEUE2 = "demo-topic-queue2";
    public static final String DEMO_TOPIC_ROUTING_KEY1 = "topic.man";
    public static final String DEMO_TOPIC_ROUTING_KEY2 = "topic.#";

    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(DEMO_TOPIC_EXCHANGE, true, false);
    }

    @Bean
    public Queue topicQueue1() {
        return new Queue(DEMO_TOPIC_QUEUE1, true);
    }

    @Bean
    public Queue topicQueue2() {
        return new Queue(DEMO_TOPIC_QUEUE2, true);
    }

    @Bean
    public Binding topicBinding1() {
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with(DEMO_TOPIC_ROUTING_KEY1);
    }

    @Bean
    public Binding topicBinding2() {
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with(DEMO_TOPIC_ROUTING_KEY2);
    }

    /**************************************************************
     * 扇形交换机
     **************************************************************/
    public static final String DEMO_FANOUT_EXCHANGE = "demo-fanout-exchange";
    public static final String DEMO_FANOUT_QUEUE1 = "demo-fanout-queue1";
    public static final String DEMO_FANOUT_QUEUE2 = "demo-fanout-queue2";

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(DEMO_FANOUT_EXCHANGE, true, false);
    }

    @Bean
    public Queue fanoutQueue1() {
        return new Queue(DEMO_FANOUT_QUEUE1, true);
    }

    @Bean
    public Queue fanoutQueue2() {
        return new Queue(DEMO_FANOUT_QUEUE2, true);
    }

    @Bean
    public Binding fanoutBinding1() {
        return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
    }

    @Bean
    public Binding fanoutBinding2() {
        return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
    }

    /**************************************************************
     * 死信交换机
     **************************************************************/
    public static final String DEMO_DEAD_LETTER_DIRECT_EXCHANGE = "demo-dead-letter-direct-exchange";
    public static final String DEMO_DEAD_LETTER_QUEUE1 = "demo-dead-letter-queue1";
    public static final String DEMO_DEAD_LETTER_QUEUE2 = "demo-dead-letter-queue2";
    public static final String DEMO_DEAD_LETTER_ROUTING_KEY1 = "dl.key1";
    public static final String DEMO_DEAD_LETTER_ROUTING_KEY2 = "dl.key2";

    @Bean
    public DirectExchange deadLetterDirectExchange() {
        return new DirectExchange(DEMO_DEAD_LETTER_DIRECT_EXCHANGE, true, false);
    }

    @Bean
    public Queue deadLetterQueue1() {
        return new Queue(DEMO_DEAD_LETTER_QUEUE1, true);
    }

    @Bean
    public Queue deadLetterQueue2() {
        return new Queue(DEMO_DEAD_LETTER_QUEUE2, true);
    }

    @Bean
    public Binding deadLetterBinding1() {
        return BindingBuilder.bind(deadLetterQueue1()).to(deadLetterDirectExchange()).with(DEMO_DEAD_LETTER_ROUTING_KEY1);
    }

    @Bean
    public Binding deadLetterBinding2() {
        return BindingBuilder.bind(deadLetterQueue2()).to(deadLetterDirectExchange()).with(DEMO_DEAD_LETTER_ROUTING_KEY2);
    }

    /**************************************************************
     * RabbitTemplate配置，消息发送确认回调
     **************************************************************/
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //数据转换为json存入消息队列
        //rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        rabbitTemplate.setMandatory(true);
        //发布者/生产者消息发送到服务器交换机确认回调。注意：开启事务，需要关闭发送确认回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            log.info("消息发送到交换机确认回调-相关数据："+correlationData);
            log.info("消息发送到交换机确认回调-确认情况："+ack);
            log.info("消息发送到交换机确认回调-原因："+cause);
        });
        //开启发送事务
        //rabbitTemplate.setChannelTransacted(true);

        //消息从服务器交换机发送到队列确认回调（注意：失败时触发）
        rabbitTemplate.setReturnsCallback(returned -> {
            log.info("消息发送到队列确认回调-状态码：" + returned.getReplyCode());
            log.info("消息发送到队列确认回调-错误：" + returned.getReplyText());
            log.info("消息发送到队列确认回调-消息：" + returned.getMessage());
            log.info("消息发送到队列确认回调-交换机：" + returned.getExchange());
            log.info("消息发送到队列确认回调-路由键：" + returned.getRoutingKey());
        });

        return rabbitTemplate;
    }

//    /**
//     * 开启rabbitmq事务机制需要
//     *
//     * @param connectionFactory 连接工厂
//     * @return
//     */
//    @Bean("rabbitTransactionManager")
//    public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {
//        return new RabbitTransactionManager(connectionFactory);
//    }

//    /**
//     * rabbitmq编码监听，全局监听，监听所有队列
//     * 优先级高于注解式监听
//     *
//     * @param connectionFactory 连接
//     * @return
//     */
//    @Bean
//    public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory) {
//        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
//        container.setConnectionFactory(connectionFactory);
//        container.setQueueNames("demo-queue");
//        container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
//            System.out.println("----接收到消息----");
//            System.out.println(message.getMessageProperties());
//            System.out.println(new String(message.getBody()));
//        });
//        return container;
//    }

}
